<?php

declare(strict_types=1);

namespace App\Exchange\Process;

use App\Exchange\Job\SendTradeTickerJob;
use Carbon\Carbon;
use Hyperf\Process\AbstractProcess;
use Hyperf\Process\Annotation\Process;
use Hyperf\Redis\Redis;
use Hyperf\WebSocketClient\Client;
use Hyperf\WebSocketClient\ClientFactory;
use Hyperf\WebSocketClient\Frame;

#[Process(name: 'BinanceTradeProcess')]
class BinanceTradeProcess extends AbstractProcess
{
    public array $pair = [];
    public Client $client;
    public int $start_at;
    public Redis $redis;

    public function stop(): bool
    {
        return $this->client->close();
    }

    public function isEnable($server): bool
    {
        return false;
        return env('APP_ENV') == 'prod';
    }

    public function handle(): void
    {
        $this->redis = container()->get(Redis::class);
        $this->start();
    }

    public function initClient(): void
    {
        $this->start_at = Carbon::now()->getTimestampMs();
        $this->pair = $this->redis->hGetAll('pair');
        $list = [];
        $interval = ['1s', '1m', '5m', '15m', '30m', '1h', '4h', '12h', '1d', '1w', '1M'];
        foreach ($this->pair as $v) {
            foreach (explode(',', strtolower($v)) as $symbol) {
                $list[] = $symbol . '@ticker';//币种24hr 1s 一次
//                $list[] = $symbol . '@trade';//实时成交
                $list[] = $symbol . '@bookTicker';//按Symbol的最优挂单信息
                foreach ($interval as $in) $list[] = $symbol . '@kline_' . $in;
                break;
            }
        }
        $host = 'wss://stream.binance.com/ws/btcusdt@depth5@1000ms';
        $this->client = container()->get(ClientFactory::class)->create($host, false);
        $subscribe = [
            'method' => 'SUBSCRIBE',
            'params' => array_splice($list, 0, 1020),
            'id' => time()
        ];
        $this->client->push(json_encode($subscribe));
    }

    public function formatData(): void
    {
        while (true) {
            /** @var $msg Frame */
            $msg = $this->client->recv(1);
            if (!$msg instanceof Frame) continue;
            $res = json_decode($msg->data, true);
            if (isset($res['e'])) {
                if ($res['e'] == 'trade') {
                    $this->redis->hSet('trade', $res['s'], implode(',', [$res['p'], $res['q']]));
                } elseif ($res['e'] == '24hrTicker') {
                    //24小时价格变化（百分比）, 平均价格, 整整24小时之前，向前数的最后一次成交价格, 最新成交价格, 最新成交交易的成交量, 目前最高买单价, 目前最高买单价的挂单量, 目前最低卖单价, 目前最低卖单价的挂单量, 整整24小时前，向后数的第一次成交价格, 24小时内最高成交价, 24小时内最低成交加, 24小时内成交量, 24小时内成交额,
                    $this->redis->hSet('ticker', $res['s'], implode(',', [$res['P'], $res['w'], $res['x'], $res['c'], $res['Q'], $res['b'], $res['B'], $res['a'], $res['A'], $res['o'], $res['h'], $res['l'], $res['v'], $res['q']]));
                    fastapp_queue_push(new SendTradeTickerJob($msg->data));
                    if ($res['E'] - $this->start_at > 7200000) {
                        $this->stop();
                        break;
                    }
                } elseif ($res['e'] == 'kline') {
                    $k = 'kline_' . strtolower($res['s'] . '_' . $res['k']['i']);
                    $this->redis->publish($k, $msg->data);
                    $this->redis->rPush('ex:' . $k, $msg->data);
                    $this->redis->ltrim('ex:' . $k, -300, -1);
                }
            } else {
                if (isset($res['s'])) {
                    //买单最优挂单价格,买单最优挂单数量,卖单最优挂单价格,卖单最优挂单数量
                    $this->redis->hSet('bookTicker', $res['s'], implode(',', [$res['b'], $res['B'], $res['a'], $res['A']]));
                }
            }
        }
    }

    public function start(): void
    {
        $this->initClient();
        $this->formatData();
        $this->start();
    }
}
